package com.boarsoft.message.store.jdbc;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;
import javax.sql.DataSource;

import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.core.MessageContext;
import com.boarsoft.message.core.MessageCore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.boarsoft.common.Util;
import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.bean.MessageEntity;
import com.boarsoft.message.bean.RecipientStatus;
import com.boarsoft.message.store.MessageRoller;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

/**
 * 基于文件的
 * 
 * @author Mac_J
 *
 */
public abstract class MsgJdbcRoller implements MessageRoller {
	private static final Logger log = LoggerFactory.getLogger(MsgJdbcRoller.class);

	// protected String driver;
	// protected String url;
	// protected String user;
	// protected String password;
	/** */
	protected DataSource dataSource;

	protected ThreadLocal<Connection> connHolder = new ThreadLocal<Connection>();

	protected MsgMainTM mainTm;

	protected MsgDetailTM detailTm;

	/** 表名 */
	protected String target;

	@Autowired
	protected MessageCore messageCore;
	@Autowired
	protected MessageContext messageContext;

	public Connection getConnection() throws Exception {
		return dataSource.getConnection();
	}

	/**
	 * 防止上一次意外中断导致无法加载那些上一次有加载但未成功处理的消息
	 */
	@PostConstruct
	public void init() throws Exception {
		// 数据初始化
		Connection conn = this.getConnection();
		try {
			if (mainTm.exists(conn)) {
				mainTm.reset(conn);
			} else {
				mainTm.create(conn);
			}
			if (detailTm.exists(conn)) {
				// detailTm.reset(conn);
			} else {
				detailTm.create(conn);
			}
		} finally {
			JdbcUtil.close(conn);
		}
		mainTm.init();
		detailTm.init();
	}

	public void recreate() throws Exception {
		Connection conn = this.getConnection();
		try {
			if (mainTm.exists(conn)) {
				mainTm.drop(conn);
			}
			if (detailTm.exists(conn)) {
				detailTm.drop(conn);
			}
		} finally {
			JdbcUtil.close(conn);
		}
		this.create();
	}

	protected void create() throws Exception {
		Connection conn = this.getConnection();
		try {
			mainTm.create(conn);
			detailTm.create(conn);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	@Override
	public String append(Message m) throws Exception {
		String id = m.getId();
		if (Util.strIsEmpty(id)) {
			id = RandomUtil.genUUID();
			m.setId(id);
		}
		//Connection conn = this.getConnection();
		Connection conn = DataSourceUtils.getConnection(dataSource);
	//	try {
		conn.setAutoCommit(false);
		try {
			mainTm.insert(conn, m);
			Map<String, RecipientStatus> rm = m.getRecipients();
			if (rm.size() > 1) {
				for (String recipient : rm.keySet()) {
					detailTm.insert(conn, id, recipient, RecipientStatus.PENDING, "");
				}
			}
			//conn.commit();
		} catch (Exception e) {
			log.error("Error on perisit message {} with status {}", //
					id, m.getStatus(), e);
			return id;
		} finally {
			conn.setAutoCommit(false);
		}
//		} finally {
//			JdbcUtil.close(conn);
//		}
		// 20230724 增加事务同步器在事务提交后进行消息入队
		TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter(){
			public void afterCompletion(int status) {
				if(status==0) {
					try {
						TargetMQ tq = messageContext.getTargetQueue(m.getTarget());
						messageCore.handle(tq,m);
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
		});

		return id;
	}

	@Override
	public long update(Message m, String memo) throws Exception {
		return this.update(m.getId(), m.getStatus(), m.getTried(), memo);
	}

	@Override
	public long update(String id, short status, int tried, String memo) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.update(conn, id, status, tried, memo);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	@Override
	public long update(String messageId, String recipient, int tried, short status, String memo) throws Exception {
		Connection conn = this.getConnection();
		try {
			return detailTm.update(conn, messageId, recipient, tried, status, memo);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 从数据库查询指定ID的记录
	 */
	@Override
	public MessageEntity read(String id) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.get(conn, id);
			// } catch (Exception e) {
			// log.error("Error on read message {}", id, e);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	@Override
	public List<MessageEntity> reload(Date begin, Date end, short status, int limit) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.reload(this, conn, begin, end, status, limit);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	@Override
	public int truncate(Date before) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.truncate(conn, before);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 统计失效消息数量
	 * 
	 * @param beforeTime
	 *            当前时间往前推N小时的时间
	 * @return 失效消息数量
	 * @throws Exception
	 */
	public int expiredNum(Date beforeTime, short... statuses) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.expiredNum(conn, beforeTime, statuses);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 检查备份表是否已经存在
	 * 
	 * @return 存在则返回表名，不存在则返回null
	 * @throws SQLException
	 */
	public String backupExists() throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.backupExists(conn);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 统计备份表数据的数量
	 * 
	 * @param backup
	 *            备份表名
	 * @return 备份表数据数量
	 * @throws Exception
	 */
	public int backupDataNum(String backup) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.backupDataNum(conn, backup);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 创建备份表
	 * 
	 * @param currentTimeStr
	 *            当前时间字符串
	 * @throws Exception
	 */
	public void backupCreate(String currentTimeStr) throws Exception {
		Connection conn = this.getConnection();
		try {
			String tableName = mainTm.getBackupTabName(currentTimeStr);
			mainTm.create(conn, mainTm.prepareCreateTbl(tableName), mainTm.prepareCreateIdx(tableName));
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 得到备份表名
	 * 
	 * @param currentTimeStr
	 * @return
	 */
	public String getBackupTabName(String currentTimeStr) {
		return mainTm.getBackupTabName(currentTimeStr);
	}

	/**
	 * 迁移数据到备份表
	 * 
	 * @param backup
	 *            备份表名
	 * @throws Exception
	 */
	public int moveExpiredData(String backup, int pageSize, long beforeTime, short... statuses) throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.moveExpiredData(conn, backup, pageSize, beforeTime, statuses);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 删除过期消息数据
	 * 
	 * @param beforeTime
	 *            过期时间
	 * @throws Exception
	 */
	public void deleteExpiredData(long beforeTime, short... statuses) throws Exception {
		Connection conn = this.getConnection();
		try {
			mainTm.deleteData(conn, beforeTime, detailTm, statuses);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 批次删除过期消息数据
	 * 
	 * @param beforeTime
	 *            过期时间
	 * @throws Exception
	 */
	public void deleteExpiredData(long beforeTime, String tableName, short... statuses) throws Exception {
		Connection conn = this.getConnection();
		try {
			mainTm.deleteData(conn, beforeTime, detailTm, tableName, statuses);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 删除已备份数据
	 * 
	 * @param beforeTime
	 *            过期时间
	 * @param backup
	 *            备份表名
	 * @throws Exception
	 */
	public void deleteBackupData(long beforeTime, String backup) throws Exception {
		Connection conn = this.getConnection();
		try {
			mainTm.delete(conn, backup, beforeTime);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 统计备份表的数量
	 * 
	 * @throws Exception
	 */
	public int countBackupTabNum() throws Exception {
		Connection conn = this.getConnection();
		try {
			return mainTm.countBackupTabNum(conn);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	/**
	 * 删除最老的备份表
	 * 
	 * @throws Exception
	 */
	public void deleteOldestBackupTab() throws Exception {
		Connection conn = this.getConnection();
		try {
			mainTm.deleteOldestBackupTab(conn);
		} finally {
			JdbcUtil.close(conn);
		}
	}

	@Override
	public void close() {
		// TODO ?
	}

	public String getTarget() {
		return target;
	}

	public void setTarget(String target) {
		this.target = target;
	}

	public DataSource getDataSource() {
		return dataSource;
	}

	public void setDataSource(DataSource dataSource) {
		this.dataSource = dataSource;
	}
}